package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

/*
*
base分流
dwd  uv
dwd  uj
dws
* */
public class DwsTrafficVcChArIsNewPageViewWindow2 {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取三个主题的数据创建流
        String uvTopic = "dwd_traffic_unique_visitor_detail";//          独立访客
        String jvdTopic = "dwd_traffic_user_jump_detail"; //             跳出用户数
        String pageTopic = "dwd_traffic_page_log";  //执行时间   流量量       会话数、页面浏览数和浏览总时长
        String groupId = "vccharisnew_pageview_window_2103a";

        // TODO: 2025/6/30   获取流  kafka流
        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> jvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(jvdTopic, groupId));
        DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId));

/*

SELECT  a,NULL,NULL,NULL,NULL   FROM  s
UNION ALL
SELECT  NULL,b,c,d,NULL FROM  s
UNION ALL
SELECT  NULL,NULL,NULL,NULL,e  FROM  s

* */
        SingleOutputStreamOperator<TrafficPageViewBean> uvMapDS = uvDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");


                return new TrafficPageViewBean(
                        null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("isNew"),
                        1l,
                        0l,
                        0l,
                        0l,
                        0l,
                        jsonObject.getLong("ts")
                );
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> jvMapDS = jvDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");


                return new TrafficPageViewBean(
                        null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("isNew"),
                        0l,
                        0l,
                        0l,
                        0l,
                        1l,
                        jsonObject.getLong("ts")
                );
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> pageMapDS = pageDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");

                Long sv=0l;
                if(page.getString("last_page_id")==null){
                    sv=1l;
                }

                return new TrafficPageViewBean(
                        null,
                        null,
                        common.getString("vc"),
                        common.getString("ch"),
                        common.getString("ar"),
                        common.getString("is_new"),
                        0l,
                        sv,
                        1l,
                        page.getLong("during_time"),
                        0l,
                        jsonObject.getLong("ts")
                );
            }
        });


        SingleOutputStreamOperator<TrafficPageViewBean> trafficPageViewBeanSingleOutputStreamOperator = pageMapDS.union(uvMapDS, jvMapDS)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(5l))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));

        //key  by   开窗   聚合reduce  输出
        SingleOutputStreamOperator<TrafficPageViewBean> reduce = trafficPageViewBeanSingleOutputStreamOperator.keyBy(new KeySelector<TrafficPageViewBean,
                Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                return Tuple4.of(value.getVc(), value.getCh(), value.getAr(), value.getIsNew());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10l)))
                .allowedLateness(Time.seconds(15l))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());

                        return value1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                                      TimeWindow window,
                                      Iterable<TrafficPageViewBean> input,
                                      Collector<TrafficPageViewBean> out) throws Exception {
                        //聚合完的结果放在迭代器里面
                        TrafficPageViewBean pageViewBean = input.iterator().next();
                        long start = window.getStart();
                        long end = window.getEnd();
                        pageViewBean.setStt(new Timestamp(start));
                        pageViewBean.setEdt(new Timestamp(end));
                        pageViewBean.setTs(System.currentTimeMillis());
                        out.collect(pageViewBean);
                    }
                });

        reduce.print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
        reduce.addSink(MyClickHouseUtil.getSinkFunction("insert  into  gmall.dws_traffic_vc_ch_ar_is_new_page_view_window  values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        //TODO 8.启动任务
        env.execute("DwsTrafficVcChArIsNewPageViewWindow");

    }

}
